#include "StdAfx.h"
#include "ActiveMqWrapper.h"

std::string g_sTopic = "com.hikvision.mq.topic";
std::string g_sQueue = "com.hikvision.mq.queue";

CActiveMqWrapper::CActiveMqWrapper()
{
	m_pConnect = NULL;
	m_pSession = NULL;

	m_pServerDestion = NULL;
	m_pServerProducer = NULL;

	m_pClientDestion = NULL;
	m_pClientConsumer = NULL;

	m_sMqUrl = "";

	activemq::library::ActiveMQCPP::initializeLibrary();
}

CActiveMqWrapper::~CActiveMqWrapper()
{
	activemq::library::ActiveMQCPP::shutdownLibrary();
}

bool CActiveMqWrapper::Init()
{
	bool bRet = false;
	do 
	{
		m_sMqUrl = "tcp://10.21.37.17:61616";
		try
		{
			auto_ptr<cms::ConnectionFactory> pFactory(cms::ConnectionFactory::createCMSConnectionFactory(m_sMqUrl.c_str()));

			m_pConnect = pFactory->createConnection();
			if (m_pConnect == NULL)
			{
				break;
			}
			m_pConnect->start();
			m_pConnect->setExceptionListener(this);

			m_pSession = m_pConnect->createSession(cms::Session::AUTO_ACKNOWLEDGE);
			if (m_pSession == NULL)
			{
				break;
			}

			if (!CreateQueue())
			{
				break;
			}

			if (!CreateTopic())
			{
				break;
			}

			bRet = true;
		}
		catch (cms::CMSException& e)
		{
			Uninit();
			
			std::string sException = e.getMessage();
			e.printStackTrace();
		}

	} while (0);
	
	return bRet;
}

bool CActiveMqWrapper::Uninit()
{
	try
	{
		if (m_pServerDestion)
		{
			delete m_pServerDestion;
			m_pServerDestion = NULL;
		}

		if (m_pServerProducer)
		{
			delete m_pServerProducer;
			m_pServerProducer = NULL;
		}

		if (m_pClientDestion)
		{
			delete m_pClientDestion;
			m_pClientDestion = NULL;
		}

		if (m_pClientConsumer)
		{
			delete m_pClientConsumer;
			m_pClientConsumer = NULL;
		}

		if (m_pSession)
		{
			m_pSession->close();
			delete m_pSession;
			m_pSession = NULL;
		}

		if (m_pConnect)
		{
			m_pConnect->close();
			delete m_pConnect;
			m_pConnect = NULL;
		}
	}
	catch (cms::CMSException& e)
	{
		e.printStackTrace();
	}

	return true;
}

void CActiveMqWrapper::onException( const cms::CMSException& ex )
{
	ex.printStackTrace();	
}

bool CActiveMqWrapper::CreateTopic()
{
	bool bRet = false;
	do 
	{
		//m_pClientDestion = m_pSession->createTopic(g_sTopic);
		m_pClientDestion = m_pSession->createQueue(g_sQueue);
		if (m_pClientDestion == NULL)
		{
			break;
		}

		m_pClientConsumer = m_pSession->createConsumer(m_pClientDestion);
		if (m_pClientConsumer == NULL)
		{
			break;
		}
		m_pClientConsumer->setMessageListener(&m_MqListener);

		bRet = true;

	} while (0);
	
	return bRet;
}
bool CActiveMqWrapper::CreateQueue()
{
	bool bRet = false;

	do 
	{
		//m_pServerDestion = m_pSession->createTopic(g_sTopic);
		m_pServerDestion = m_pSession->createQueue(g_sQueue);
		if (m_pServerDestion == NULL)
		{
			break;
		}

		m_pServerProducer = m_pSession->createProducer(m_pServerDestion);
		m_pServerProducer->setDeliveryMode(cms::DeliveryMode::NON_PERSISTENT);

		bRet = true;

	} while (0);
	
	return bRet;
}

bool CActiveMqWrapper::SendMqMsg()
{
	cms::MapMessage* mapMsg = NULL;
	try 
	{
		if (m_pSession && m_pServerProducer)
		{
			mapMsg = m_pSession->createMapMessage();
			mapMsg->setInt("Type", 1);
			mapMsg->setInt("Id", 2);
			mapMsg->setString("Name", stdListen::mb2u8("wyt"));
			m_pServerProducer->send(mapMsg);
		}
	}
	catch ( cms::CMSException& e ) 
	{
		e.printStackTrace();
	}

	if (mapMsg)
	{
		delete mapMsg;
		mapMsg = NULL;
	}

	return true;
}